package com.xzx.flink.streamapi.processfunction;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import com.xzx.flink.streamapi.window.UrlViewCount;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/4 15:28
 */
public class TopN_02_KeyedProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1. 先计算出每个url的pv数
        SingleOutputStreamOperator<UrlViewCount> aggregate = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                            @Override
                            public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }))
                .keyBy(ClickEvent::getUrl)
                .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.seconds(5)))
                .aggregate(new AggregateFunction<ClickEvent, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(ClickEvent value, Long accumulator) {
                        return accumulator + 1;
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return a + b;
                    }
                }, new ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow>() {
                    @Override
                    public void process(String key, ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<UrlViewCount> out) throws Exception {
                        // 这个会被5秒执行一次
                        UrlViewCount urlViewCount = new UrlViewCount();
                        urlViewCount.setWindowStart(context.window().getStart());
                        urlViewCount.setWindowEnd(context.window().getEnd());
                        urlViewCount.setCount(elements.iterator().next());
                        urlViewCount.setUrl(key);
                        out.collect(urlViewCount);
                    }
                });
        // 2. 再计算topN
        aggregate.keyBy(UrlViewCount::getWindowEnd)// 同一个结束窗口的数据放到一个分组里面
                .process(new KeyedProcessFunction<Long, UrlViewCount, String>() {

                    private ListState<UrlViewCount> urlViewCountListState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        //初始化一个状态
                        urlViewCountListState = getRuntimeContext().getListState(new ListStateDescriptor<UrlViewCount>("url-view-count-list", Types.POJO(UrlViewCount.class)));
                    }

                    @Override
                    public void processElement(UrlViewCount value, KeyedProcessFunction<Long, UrlViewCount, String>.Context ctx, Collector<String> out) throws Exception {
                        // 来一条，往状态里面存储一下
                        urlViewCountListState.add(value);
                        // 注册一个定时器，窗口关闭时触发
                        ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<Long, UrlViewCount, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        List<UrlViewCount> list = new ArrayList<>();
                        for (UrlViewCount urlViewCount : urlViewCountListState.get()) {
                            list.add(urlViewCount);
                        }

                        List<UrlViewCount> sortList = list.stream().sorted((o1, o2) -> o2.getCount().compareTo(o1.getCount())).collect(Collectors.toList());
                        for (int i = 0; i < 2 && i < sortList.size(); i ++) {
                            UrlViewCount urlViewCount = sortList.get(i);
                            String info = "浏览量No." + (i + 1) + " url:" + urlViewCount.getUrl() + ";浏览量:" + urlViewCount.getCount() + ";窗口结束时间：" + new Timestamp(urlViewCount.getWindowEnd());
                            out.collect(info);
                        }
                        out.collect("============分割线====================");

                        urlViewCountListState.clear();
                    }
                })
                .print();

        env.execute(TopN_02_KeyedProcessFunction.class.getSimpleName());
    }
}
